package jtcl.tech.kafka.factory;

import ch.qos.logback.core.joran.util.beans.BeanUtil;
import jtcl.tech.kafka.annotation.KafkaConsumerListener;
import jtcl.tech.kafka.dto.KafkaConfigDTO;
import jtcl.tech.kafka.listener.ConsumerBatchListener;
import jtcl.tech.kafka.listener.KafkaInitListener;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.BeanUtils;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <p>Kafka 监听工厂</p>
 *
 * @author sjt
 * @since 2021-08-05 16:37
 */
@Slf4j
public class KafkaFactory implements ApplicationContextAware{
    /**
     * 缓存消费者监听对象
     */
    protected static final ConcurrentHashMap<String, ConcurrentMessageListenerContainer> CONTAINER_MAP =
            new ConcurrentHashMap<>();
    /**
     * <p>kafka 配置对象</p>
     * <p>优先级：kafkaConfigOfGlobal < @KafkaConsumerListener < config(kafkaConfigDTO) </p>
     */
    public static final KafkaConfigDTO KAFKA_CONFIG_OF_GLOBAL = new KafkaConfigDTO();

    @Setter
    ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory;
    @Setter
    KafkaProperties kafkaProperties;
    @Setter
    KafkaTemplate kafkaTemplate;

    @Setter
    KafkaProducerFactory producerFactory;
    @Setter
    KafkaConsumerFactory consumerFactory;

    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        KafkaFactory.applicationContext = applicationContext;
    }


    public void init() {
        KafkaInitListener kafkaInitListener = applicationContext.getBean(KafkaInitListener.class);
        if (Objects.nonNull(kafkaInitListener)) {
            kafkaInitListener.updateConfig(KAFKA_CONFIG_OF_GLOBAL);
        }

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    resetKafkaProducer();
                    startupConsumerListener();
                    System.out.println("完成kafka初始化");
                } catch (Exception e) {
                    log.error("队列初始化异常", e);
                }
            }
        });
        thread.setName("kafkaFactory");
        thread.start();
    }

    public void resetKafkaProducer() {
        if (StringUtils.isBlank(KAFKA_CONFIG_OF_GLOBAL.getBootstrapServers())
                && CollectionUtils.isEmpty(kafkaProperties.getBootstrapServers())) {
            log.info("未设置BootstrapServers,无法创建Producer");
            return;
        }
        if (StringUtils.isNotBlank(KAFKA_CONFIG_OF_GLOBAL.getBootstrapServers())) {
            kafkaProperties.setBootstrapServers(
                    Arrays.asList(KAFKA_CONFIG_OF_GLOBAL.getBootstrapServers().split(","))
            );
        }
        Map configs = producerFactory.getConfigs();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        if (producerFactory.getProducer() != null) {
            producerFactory.destroy();
        }
        kafkaTemplate.flush();
    }

    public void startupConsumerListener() {
        log.info("开始初始化kafka 消费者");
        CONTAINER_MAP.values().forEach(AbstractMessageListenerContainer::stop);

        Map<String, ConsumerBatchListener> beansOfType = applicationContext
                .getBeansOfType(ConsumerBatchListener.class);
        beansOfType.values().forEach(bean -> {
            //获取配置
            KafkaConfigDTO kafkaConfigDTO = getKafkaConfigDTO(bean);
            if (StringUtils.isBlank(kafkaConfigDTO.getName())) {
                kafkaConfigDTO.setName(bean.getClass().getSimpleName());
            }

            if (StringUtils.isBlank(kafkaConfigDTO.getBootstrapServers())) {
                log.info("消费者{}未设置BootstrapServers", bean.getClass().getName());
                return;
            }
            if (kafkaConfigDTO.getTopics().length == 0) {
                log.info("消费者{}未设置Topics", bean.getClass().getName());
                return;
            }

            kafkaProperties.setBootstrapServers(
                    Arrays.asList(kafkaConfigDTO.getBootstrapServers().split(","))
            );
            KafkaConsumerFactory consumerFactory = (KafkaConsumerFactory) applicationContext.getBean(ConsumerFactory.class);
            Map configs = consumerFactory.getConfigs();
            configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());

            ConcurrentMessageListenerContainer<?, ?> container = containerFactory.createContainer(kafkaConfigDTO.getTopics());
            ContainerProperties containerProperties = container.getContainerProperties();
            if (StringUtils.isNotBlank(kafkaConfigDTO.getGroupId())) {
                containerProperties.setGroupId(kafkaConfigDTO.getGroupId());
            }
            container.setBeanName(kafkaConfigDTO.getName());
            container.setupMessageListener(bean);
            container.start();

            CONTAINER_MAP.put(kafkaConfigDTO.getName(), container);
            log.info("开始监听，topic={},groupId={}", kafkaConfigDTO.getTopics(), container.getGroupId());
        });
        log.info("监听初始化完成，启动监听数量：" + beansOfType.size());
    }

    public void stopConsumerListener(String name) {
        ConcurrentMessageListenerContainer container = CONTAINER_MAP.get(name);
        container.stop();
        CONTAINER_MAP.remove(name);
        log.info("成功关闭消费者：{},topics={}", name, container.getContainerProperties().getTopics());
    }

    public void stopAllConsumerListener() {
        CONTAINER_MAP.forEach((name, container) -> {
            container.stop();
            log.info("成功关闭消费者：{},topics={}", name, container.getContainerProperties().getTopics());
        });
        CONTAINER_MAP.clear();
        log.info("所有消费者全都关闭");
    }

    /**
     * <p> 配置参数 </p>
     *
     * @param bean
     * @return cec.ics.common.queue.kafka.KafkaConfigDTO
     * @author 施锦庭
     * @since 2021-08-12 16:06
     **/
    private KafkaConfigDTO getKafkaConfigDTO(ConsumerBatchListener bean) {
        KafkaConfigDTO kafkaConfigDTO=new KafkaConfigDTO();
        BeanUtils.copyProperties(KAFKA_CONFIG_OF_GLOBAL, kafkaConfigDTO);
        KafkaConsumerListener kafkaConsumerListener = bean.getClass().getAnnotation(KafkaConsumerListener.class);
        if (Objects.nonNull(kafkaConsumerListener)) {
            if (StringUtils.isNotBlank(kafkaConsumerListener.name())) {
                kafkaConfigDTO.setName(kafkaConsumerListener.name());
            }
            if (StringUtils.isNotBlank(kafkaConsumerListener.bootstrapServers())) {
                kafkaConfigDTO.setBootstrapServers(kafkaConsumerListener.bootstrapServers());
            }

            if (StringUtils.isNotBlank(kafkaConsumerListener.groupId())) {
                kafkaConfigDTO.setGroupId(kafkaConsumerListener.groupId());
            }
            if (kafkaConsumerListener.topics().length != 0) {
                kafkaConfigDTO.setTopics(kafkaConsumerListener.topics());
            }
        }
        bean.config(kafkaConfigDTO);
        return kafkaConfigDTO;
    }


}
